package com.liu.demo.producer;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.util.DigestUtils;

import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 * Kafka生产者
 */
public class CustomProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        // 配置可以使用具体的字符串，也可以使用producerConfig中配置的静态变量名
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop-01:9092,hadoop-02:9092,hadoop-03:9092");
        // ack机制
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        // 重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, 1);
        // 批次大小:消息大小为16384才发送消息
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // 等待时间:如果消息大小迟迟不为batch.size大小，则等待linger.ms时间后直接发送
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        // ReadAccumulator缓冲区大小
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        // 序列化  org.apache.kafka.common.serialization.StringSerializer
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        //  自定义分区器
//        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.example.kafka.partition.CustomPartitioner");


        // 构造producer
        Producer<String, String> producer = new KafkaProducer<String, String>(props);


        String topic = "TXHKP_YXSJ_MES";
        JSONObject object = new JSONObject();
        object.put("operate","delete");
        String YT = "1ec56485-9f35-471a-ae62-42a07dec8c11";

        JSONObject entity = new JSONObject();
        // 生产消息
        for (int i = 0; i < 100; i++) {

            entity.put("station_code",i);
            entity.put("data_source","01");
            entity.put("station_name","测试-" +i);
            entity.put("device_code",i);

            String secret = YT + entity.getString("station_code")+ "#"
                    +entity.getString("device_code")+ "#"
                    +entity.getString("data_source");
            secret = DigestUtils.md5DigestAsHex(secret.getBytes(StandardCharsets.UTF_8));
            object.put("secret",secret);
            object.put("entity",entity);

            // 构造消息体
            producer.send(new ProducerRecord<>(topic, object.toString()));
        }
        producer.close();
    }
}
